package com.hw.nio.netty.redis;


import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.CodecException;
import io.netty.handler.codec.redis.*;
import io.netty.util.CharsetUtil;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * Redis 协议实现
 */
public class RedisClient {

    private Channel channel;

    /**
     * 实现步骤
     *  1、使用bootStrap
     *      1、配置事件监听组
     *      2、配置channel管道
     *      3、配置handler 初始化channel 添加事件处理器：编码、解码、读写处理、数据聚合
     * @param args
     * @throws InterruptedException
     * @throws IOException
     */

    public static void main(String[] args) throws InterruptedException, IOException {
        RedisClient redisClient = new RedisClient();
        redisClient.openServer();

        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));
        while (true) {
            String line = bufferedReader.readLine();
            System.out.print(">");
            redisClient.channel.writeAndFlush(line);
        }

    }

    public void openServer() throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(new NioEventLoopGroup(1))
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast("decode", new RedisDecoder());
                        ch.pipeline().addLast("bulk-aggregator",new RedisBulkStringAggregator());
                        ch.pipeline().addLast("array-aggregator",new RedisArrayAggregator());
                        ch.pipeline().addLast("encode", new RedisEncoder());
                        ch.pipeline().addLast("dispatcher", new MyServlet());
                    }
                });
        channel = bootstrap.connect(new InetSocketAddress("127.0.0.1", 6379)).sync().channel();

    }

    class MyServlet extends ChannelDuplexHandler {
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            if (!(msg instanceof String)) {
                ctx.write(msg);
            }
            // 按空格切分命令
            String[] commands = ((String) msg).split("\\s+");
            List<RedisMessage> list = new ArrayList<>(commands.length);
            for (String command : commands) {
                //将每一个命令传入集合中
                list.add(new FullBulkStringRedisMessage(Unpooled.wrappedBuffer(command.getBytes(StandardCharsets.UTF_8))));
            }
            // 将单个redis命令封装至redis命令集合
            ArrayRedisMessage arrayRedisMessage = new ArrayRedisMessage(list);
            // 向服务器写入命令
            ctx.write(arrayRedisMessage, promise);
        }

        // 读取redis服务返回的消息
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            printAggregatedRedisResponse((RedisMessage) msg);
        }

        private void printAggregatedRedisResponse(RedisMessage msg) {
            if (msg instanceof SimpleStringRedisMessage) {
                System.out.println(((SimpleStringRedisMessage) msg).content());
            } else if (msg instanceof ErrorRedisMessage) {
                System.out.println(((ErrorRedisMessage) msg).content());
            } else if (msg instanceof IntegerRedisMessage) {
                System.out.println(((IntegerRedisMessage) msg).value());
            } else if (msg instanceof FullBulkStringRedisMessage) {
                System.out.println(getString((FullBulkStringRedisMessage) msg));
            } else if (msg instanceof ArrayRedisMessage) {
                for (RedisMessage child : ((ArrayRedisMessage) msg).children()) {
                    printAggregatedRedisResponse(child);
                }
            } else {
                throw new CodecException("unknown message type: " + msg);
            }
        }

        private String getString(FullBulkStringRedisMessage msg) {
            if (msg.isNull()) {
                return "(null)";
            }
            return msg.content().toString(CharsetUtil.UTF_8);
        }
    }
}
